package com.example.grpc.service.helloworld;

import com.google.common.collect.Lists;
import io.grpc.examples.hellostreaming.DubboStreamingGreeterGrpc;

import io.grpc.examples.hellostreaming.HelloReply;
import io.grpc.examples.hellostreaming.HelloRequest;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.dubbo.config.annotation.DubboService;

import java.util.List;
import java.util.Random;

@Slf4j
@DubboService(interfaceClass = DubboStreamingGreeterGrpc.IStreamingGreeter.class, loadbalance = "roundrobin", version = "1.0")
public class StreamingGreeterServiceImpl extends DubboStreamingGreeterGrpc.StreamingGreeterImplBase {

    /**
     * 请求一次返回10次
     * @param request
     * @param responseObserver
     */
    @Override
    public void sayHelloTen(HelloRequest request, StreamObserver<HelloReply> responseObserver) {
        String name = request.getName();
        for (int i = 1; i < 11; i++) {
            HelloReply build = HelloReply.newBuilder().setMessage("hello " + name + " --" + i).build();
            responseObserver.onNext(build);
        }
        responseObserver.onCompleted();
    }

    /**
     * 每请求一次返回一次
     * @param responseObserver
     * @return
     */
    @Override
    public StreamObserver<HelloRequest> sayHelloStreaming(final StreamObserver<HelloReply> responseObserver) {
        return new StreamObserver<HelloRequest>() {
            int count = 1;

            @Override
            public void onNext(HelloRequest helloRequest) {
                HelloReply build = HelloReply.newBuilder().setMessage("hello " + helloRequest.getName() + " --" + count).build();
                responseObserver.onNext(build);
                count++;
            }

            @Override
            public void onError(Throwable throwable) {
                log.warn("routeChat cancelled");
            }

            @Override
            public void onCompleted() {
                responseObserver.onCompleted();
            }
        };
    }

    /**
     * 请求多次，从中随机选择一个操作返回
     * @param responseObserver
     * @return
     */
    @Override
    public StreamObserver<HelloRequest> sayHelloStreamingGetOneReply(final StreamObserver<HelloReply> responseObserver) {
        return new StreamObserver<HelloRequest>() {
            List<String> reqList = Lists.newArrayList();

            int count = 1;

            @Override
            public void onNext(HelloRequest helloRequest) {
                log.info("收到第{}个数据", count);
                reqList.add("hello " + helloRequest.getName() + " --" + count);
                count++;
            }

            @Override
            public void onError(Throwable throwable) {
                log.warn("routeChat cancelled");
            }

            @Override
            public void onCompleted() {
                Random random = new Random();
                String str = reqList.get(random.nextInt(reqList.size()));
                HelloReply build = HelloReply.newBuilder().setMessage(str).build();
                // 只有一个返回，所以执行一次onNext,执行多次后面的会被拦截
                // Cancelling the stream with status Status{code=INTERNAL, description=Too many responses, cause=null}
                responseObserver.onNext(build);
                responseObserver.onCompleted();
            }
        };
    }
}
